package org.liang.code.jms.queue;


import jakarta.jms.JMSConsumer;
import jakarta.jms.JMSContext;
import jakarta.jms.JMSException;
import jakarta.jms.Session;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory;
import org.liang.code.utils.jms.JMSCommons;

import java.util.concurrent.CountDownLatch;

/**
 * 队列消费者
 */
@Slf4j
public class TConsumer implements Runnable{

    private final CountDownLatch latch;
    private final ActiveMQJMSConnectionFactory factory;
    private final Boolean transactionMessage;

    public TConsumer(CountDownLatch latch, ActiveMQJMSConnectionFactory factory, Boolean transactionMessage) {
        this.latch = latch;
        this.factory = factory;
        this.transactionMessage = transactionMessage;
    }

    @SneakyThrows
    @Override
    public void run() {

        JMSContext context = null;
        JMSConsumer consumer = null;
        try {
            // 创建JMSContext
            if (transactionMessage) {
                context = factory.createContext(Session.SESSION_TRANSACTED);
            }else {
                context = factory.createContext(Session.AUTO_ACKNOWLEDGE);
            }

            // 创建消费者
            consumer = context.createConsumer(context.createQueue(JMSCommons.CORE_QUEUE));
            // 读取消息
            JMSCommons.readMessage(consumer);
            // 使用事务
            if (transactionMessage) {
                context.commit();
            }
            Thread.sleep(100);
        } catch (JMSException e) {
            if (transactionMessage) {
                // 回滚事务
                context.rollback();
                log.error("事务已回滚！", e);
            }
        } finally {
            // 关闭消费者
            if (consumer != null) {
                consumer.close();
            }
            // 关闭上下文
            if (context != null) {
                context.close();
            }
            latch.countDown();
        }
    }


}
